<?php
 
namespace app\rabbitmq\controller;
 
 
use think\Config;
use think\Log;
use think\Db;
use think\Cache;
use think\Exception;
//引入rabbitMq所需类
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
 
class MqConsumer
{
    /**
     * 消费者(路由模式)
     * @param $name 配置文件下rabbitmq文件前缀名
     * @throws \ErrorException
     */
    public function consumer($name)
    {
        //获取rabbitMq配置
        $amqp = Config::get('AMQP');
        $amqpDefail = Config::get('' . $name . '_queue');

        //建立连接
        $connection = new AMQPStreamConnection(
            $amqp['host'],
            $amqp['port'],
            $amqp['username'],
            $amqp['password']
        );
 
        //建立通道
        $channel = $connection->channel();
 
        //流量控制(也被称为公平调度，这里设置一个消费者一次只处理200条消息)
        $num = 100;
        if ($name == 'ffmpeg') {
            $num = 1;
        }
        $channel->basic_qos(null, $num, null);
 
        //初始化交换机
        $channel->exchange_declare($amqpDefail['exchange_name'], $amqpDefail['exchange_type'], false, true, false);
 
        //初始化队列
        $channel->queue_declare($amqpDefail['queue_name'], false, true, false, false);
 
        //绑定队列与交换机
        $channel->queue_bind($amqpDefail['queue_name'], $amqpDefail['exchange_name'], $amqpDefail['route_key']);
        
        //消费消息
        $channel->basic_consume($amqpDefail['queue_name'], $amqpDefail['consumer_tag'], false, false, false, false, [$this, 'callback']);
        
        //注册退出函数行为(其实用不到,因为要保证消费者一直运行,所以不能断开连接))
        // register_shutdown_function([$this,'shutdown'], $channel, $connection);
 
        //消息未处理完毕时,循环监听并一直处理上方callback方法的逻辑
        $channel->consume();
 
    }
    
    /**
     * 回调后消息处理(业务逻辑放置此处)
     * @param  $msg
     */
    public function callback($msg)
    {
        $data = json_decode($msg->body,true);//拿到数据
        try {
            switch (str_replace(getenv('APP_NAME_EN').'_', '', $msg->delivery_info['consumer_tag'])) {
                case 'curl':
                    sendcurl($data['url'], $data['data'], $data['type']);
                    break;
                case 'action':
                    action($data['action'], $data['data']);
                    foreach ($data['data'] as $key => $value) {
                        if ($key == 'timer_task_id') {
                            Db::name('timer_task')->where(['id'=>$value])->update(['run_time'=>datetime()]);
                        }
                    }
                    break;
                case 'class_event':
                    if (!class_exists($data['class'])) {
                        Exception('类不存在', 100032);
                    }
                    $class = new $data['class'];
                    $method = $data['method'];
                    $params = $data['params'] ?? [];
                    if (method_exists($class, $method)) {
                        $class::$method(...$params);
                    } else {
                        Exception('函数不存在', 100032);
                    }
                    break;
                case 'ffmpeg':
                    $hour = date('H');
                    if ($hour > 18 || $hour < 2) {
                        Cache::store('redis')->set('rabbitMQReConsumption:'.$data['id'], $data['id'], 3600);
                    } else {
                        action('api/Tool/convertVideoToM3U8AndUpload',['id'=>$data['id']]);
                    }
                    break;
                
                default:
                    echo $msg->delivery_info['consumer_tag'];
                    break;
            }
            //如果有个值是quit,则让该消费者停止消费
            /*if($data['aa'] == 'quit'){
                $msg->getChannel()->basic_cancel($msg->getConsumerTag());
            }*/
            $msg->ack();  //消息应答：这波200条消息处理完毕后进行消息确认,告诉mq可以开始发下一波200条消息了
        } catch (\Exception $e) {
            Log::write($msg->delivery_info['consumer_tag'] . ':' . $msg->body . ':' . $e->getMessage(), 'error');
        }
    }
 
    
    
    /**
     * @param $channel 信道
     * @param \PhpAmqpLib\Connection\AbstractConnection $connection
     */
    function shutdown($channel, $connection)
    {
        $channel->close(); //关闭通道
        $connection->close(); //断开连接
    }
 
 
}